package com.ruleengine.worker.runner.rule;

import java.lang.reflect.Field;
import java.util.Queue;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import com.ruleengine.worker.exception.RuleRunnerException;
import com.ruleengine.worker.model.Rule;

import cn.hutool.core.util.ReflectUtil;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.ext.reactivestreams.ReactiveReadStream;
import io.vertx.ext.reactivestreams.ReactiveWriteStream;
import io.vertx.ext.reactivestreams.impl.ReactiveReadStreamImpl;
import io.vertx.ext.reactivestreams.impl.ReactiveWriteStreamImpl;

/**
 * Pub + Sub RuleRunner
 * https://vertx.io/docs/vertx-core/java/#streams
 * https://vertx.io/docs/vertx-reactive-streams/java/
 *
 * @author lianghanwei on 2024/2/18
 */
public abstract class PubSubRuleRunner<T, R> implements RuleRunner<T, R> {

    protected String name;
    protected Rule rule;
    private ReactiveReadStream<T> subscriber;
    private ReactiveWriteStream<R> publisher;

    public PubSubRuleRunner(Vertx vertx, String name, Rule rule) {
        this.name = name;
        this.rule = rule;
        this.subscriber = ReactiveReadStream.readStream();
        this.publisher = ReactiveWriteStream.writeStream(vertx);
    }

    /**
     * As a Publisher
     */
    @Override
    public void subscribe(Subscriber<? super R> subscriber) {
        publisher.subscribe(subscriber);
    }

    public Future<Void> write(R item) {
        return publisher.write(item);
    }

    public int getPubPendingSize() {
        Field pendingField = ReflectUtil.getField(ReactiveWriteStreamImpl.class, "pending");
        pendingField.setAccessible(true);
        try {
            Queue<R> pendingQueue = (Queue<R>) pendingField.get(publisher);
            return pendingQueue.size();
        } catch (IllegalAccessException e) {
            throw new RuleRunnerException("Reflect publisher pending size failed", e);
        }
    }

    public void close() {
        publisher.close();
    }

    /**
     * As a Subscriber
     */
    @Override
    public void onSubscribe(Subscription subscription) {
        subscriber.onSubscribe(subscription);
        subscriber.handler(item -> {
            R processedItem = process(item);

            if (processedItem != null) {
                publisher.write(processedItem);
            }

            // Flow control
            if (publisher.writeQueueFull()) {
                subscriber.pause();
                publisher.drainHandler(v -> subscriber.resume());
            }
        });
    }

    @Override
    public void onNext(T item) {
        subscriber.onNext(item);
    }

    @Override
    public void onError(Throwable t) {
        subscriber.onError(t);
    }

    @Override
    public void onComplete() {
        subscriber.onComplete();
    }

    public int getSubPendingSize() {
        Field pendingField = ReflectUtil.getField(ReactiveReadStreamImpl.class, "pending");
        pendingField.setAccessible(true);
        try {
            Queue<T> pendingQueue = (Queue<T>) pendingField.get(subscriber);
            return pendingQueue.size();
        } catch (IllegalAccessException e) {
            throw new RuleRunnerException("Reflect subscriber pending size failed", e);
        }
    }

}
